{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%run startup.py"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "application/javascript": [
       "$.getScript('./assets/js/ipython_notebook_toc.js')"
      ],
      "text/plain": [
       "<IPython.core.display.Javascript object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%javascript\n",
    "$.getScript('./assets/js/ipython_notebook_toc.js')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A Decision Tree of Observable Operators\n",
    "\n",
    "## Part 7: Meta Operations\n",
    "\n",
    "> source: http://reactivex.io/documentation/operators.html#tree.  \n",
    "> (transcribed to RxPY 1.5.7, Py2.7 / 2016-12, Gunther Klessinger, [axiros](http://www.axiros.com))  \n",
    "\n",
    "**This tree can help you find the ReactiveX Observable operator you’re looking for.**  \n",
    "See [Part 1](./A Decision Tree of Observable Operators. Part I - Creation.ipynb) for Usage and Output Instructions.  \n",
    "\n",
    "We also require acquaintance with the [marble diagrams](./Marble Diagrams.ipynb) feature of RxPy.\n",
    "\n",
    "<h2 id=\"tocheading\">Table of Contents</h2>\n",
    "<div id=\"toc\"></div>\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# I want to convert the entire sequence of items emitted by an Observable into some other data structure  [to_iterable/to_list, to_blocking, to_dict, to_future, to_marbles, to_set](http://reactivex.io/documentation/operators/to.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== to_list ==========\n",
      "\n",
      "module rx.linq.observable.tolist\n",
      "@extensionmethod(Observable, alias=\"to_iterable\")\n",
      "def to_list(self):\n",
      "    Creates a list from an observable sequence.\n",
      "\n",
      "    Returns an observable sequence containing a single element with a list\n",
      "    containing all the elements of the source sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "('got', 'a', 0.011667013168334961)\n",
      "('got', 'b', 0.22758698463439941)\n",
      "('got', 'c', 0.335313081741333)\n",
      "('got', ['a', 'b', 'c'], 0.3470189571380615)\n"
     ]
    }
   ],
   "source": [
    "rst(O.to_iterable)\n",
    "s = marble_stream(\"a--b-c|\")\n",
    "l, ts = [], time.time()\n",
    "def on_next(listed):\n",
    "    print('got', listed, time.time()-ts)\n",
    "\n",
    "for i in (1, 2):    \n",
    "    d = s.subscribe(on_next)\n",
    "    # second run: only one value, the list.\n",
    "    s = s.to_list()\n",
    "    # both are started around same time -> check time deltas\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== to_blocking ==========\n",
      "\n",
      "module rx.linq.observable.toblocking\n",
      "@extensionmethod(Observable)\n",
      "def to_blocking(self):\n",
      "    n.a.\n",
      "--------------------------------------------------------------------------------\n",
      "In some implementations of ReactiveX, there is also an operator that converts an Observable into a “Blocking” Observable. A Blocking Observable extends the ordinary Observable by providing a set of methods, operating on the items emitted by the Observable, that block. Some of the To operators are in this Blocking Obsevable set of extended operations.\n",
      "\n",
      "\n",
      "========== __iter__ ==========\n",
      "\n",
      "module rx.linq.observable.blocking.toiterable\n",
      "@extensionmethod(BlockingObservable)\n",
      "def __iter__(self):\n",
      "    Returns an iterator that can iterate over items emitted by this\n",
      "    `BlockingObservable`.\n",
      "\n",
      "    :param BlockingObservable self: Blocking observable instance.\n",
      "    :returns: An iterator that can iterate over the items emitted by this\n",
      "        `BlockingObservable`.\n",
      "    :rtype: Iterable[Any]\n",
      "--------------------------------------------------------------------------------\n",
      " 204.0     M 0\n",
      " 408.9     M 1\n",
      " 615.1     M 2\n",
      " 822.2     M 0\n",
      "1027.2     M 1\n",
      "1230.1     M 2\n",
      "\n",
      "\n",
      "========== for_each ==========\n",
      "\n",
      "module rx.linq.observable.blocking.foreach\n",
      "@extensionmethod(BlockingObservable)\n",
      "def for_each(self, action):\n",
      "    Invokes a method on each item emitted by this BlockingObservable and\n",
      "    blocks until the Observable completes.\n",
      "\n",
      "    Note: This will block even if the underlying Observable is asynchronous.\n",
      "\n",
      "    This is similar to Observable#subscribe(subscriber), but it blocks. Because\n",
      "    it blocks it does not need the Subscriber#on_completed() or\n",
      "    Subscriber#on_error(Throwable) methods. If the underlying Observable\n",
      "    terminates with an error, rather than calling `onError`, this method will\n",
      "    throw an exception.\n",
      "\n",
      "    Keyword arguments:\n",
      "    :param types.FunctionType action: the action to invoke for each item\n",
      "        emitted by the `BlockingObservable`.\n",
      "    :raises Exception: if an error occurs\n",
      "    :returns: None\n",
      "    :rtype: None\n",
      "--------------------------------------------------------------------------------\n",
      " 205.5   T28 0\n",
      " 408.5   T29 1\n",
      " 613.9   T30 2\n",
      "\n",
      "\n",
      "========== .observable -> getting async again ==========\n",
      "\n",
      "\n",
      " 615.9     M New subscription (observer 1) on stream 276608405\n",
      "\n",
      " 617.6     M New subscription (observer 2) on stream 276608405\n",
      " 821.8   T33 [next]  204.2: 0 -> observer 2   822.6   T32 [next]  206.4: 0 -> observer 1  \n",
      "\n",
      "1025.0   T34 [next]  407.4: 1 -> observer 2  1024.8   T35 [next]  408.6: 1 -> observer 1  \n",
      "\n",
      "1227.7   T36 [next]  610.1: 2 -> observer 2  1229.0   T37 [next]  612.8: 2 -> observer 1  \n",
      "\n",
      "1229.4   T37 [cmpl]  613.2: fin -> observer 1  1228.5   T36 [cmpl]  610.9: fin -> observer 2  \n",
      "\n"
     ]
    }
   ],
   "source": [
    "rst(O.to_blocking)\n",
    "ts = time.time()\n",
    "s = O.interval(200).take(3)\n",
    "sb = s.to_blocking()\n",
    "# this is instant:\n",
    "assert time.time() - ts < 0.2\n",
    "\n",
    "print('''In some implementations of ReactiveX, there is also an operator that converts an Observable into a “Blocking” Observable. A Blocking Observable extends the ordinary Observable by providing a set of methods, operating on the items emitted by the Observable, that block. Some of the To operators are in this Blocking Obsevable set of extended operations.''')\n",
    "# -> diffing dir(s) with dir(sb) we get:\n",
    "# __iter__\n",
    "# for_each\n",
    "# observable\n",
    "rst(sb.__iter__)\n",
    "for i in (1, 2):\n",
    "    # not interleaved results:\n",
    "    for it in sb:\n",
    "        log(it)\n",
    "\n",
    "rst(sb.for_each)        \n",
    "sb.for_each(log)\n",
    "\n",
    "header(\".observable -> getting async again\")\n",
    "# interleaved again:\n",
    "d = subs(sb.observable, name='observer 1')\n",
    "d = subs(sb.observable, name='observer 2')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== to_dict ==========\n",
      "\n",
      "module rx.linq.observable.todict\n",
      "@extensionmethod(Observable)\n",
      "to_dict(self, key_selector, element_selector=None)\n",
      "    Converts the observable sequence to a Map if it exists.\n",
      "\n",
      "    Keyword arguments:\n",
      "    key_selector -- {Function} A function which produces the key for the\n",
      "        Map.\n",
      "    element_selector -- {Function} [Optional] An optional function which\n",
      "        produces the element for the Map. If not present, defaults to the\n",
      "        value from the observable sequence.\n",
      "    Returns {Observable} An observable sequence with a single value of a Map\n",
      "    containing the values from the observable sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.6     M New subscription on stream 276526817\n",
      "   2.6     M [next]    0.7: {'a': 'aa', 'c': 'cc', 'b': 'bb'}\n",
      "   2.8     M [cmpl]    0.8: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.to_dict)\n",
    "d = subs(O.from_('abc').to_dict(key_selector=lambda x: x, element_selector=lambda a: '%s%s' % (a, a)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== to_future ==========\n",
      "\n",
      "module rx.linq.observable.tofuture\n",
      "@extensionmethod(Observable)\n",
      "def to_future(self, future_ctor=None):\n",
      "    Converts an existing observable sequence to a Future\n",
      "\n",
      "    Example:\n",
      "    future = rx.Observable.return_value(42).to_future(trollius.Future);\n",
      "\n",
      "    With config:\n",
      "    rx.config[\"Future\"] = trollius.Future\n",
      "    future = rx.Observable.return_value(42).to_future()\n",
      "\n",
      "    future_ctor -- {Function} [Optional] The constructor of the future.\n",
      "        If not provided, it looks for it in rx.config.Future.\n",
      "\n",
      "    Returns {Future} An future with the last value from the observable\n",
      "    sequence.\n",
      "--------------------------------------------------------------------------------\n",
      " 506.5     M emitting first\n",
      "1008.0     M emitting second\n",
      "1008.7     M future.result(): second\n"
     ]
    }
   ],
   "source": [
    "rst(O.to_future)\n",
    "def emit(obs):    \n",
    "    for ev in 'first', 'second':\n",
    "        sleep(.5)\n",
    "        log('emitting', ev)\n",
    "        obs.on_next(ev)\n",
    "    # vital for the future to get done:\n",
    "    obs.on_completed()\n",
    "    \n",
    "    \n",
    "try:\n",
    "    # required for py2 (backport of guidos' tulip stuffs, now asyncio)\n",
    "    # caution: people say this is not production ready and will never be.\n",
    "    import trollius\n",
    "    f = rx.Observable.create(emit).to_future(trollius.Future)\n",
    "    # this is async, not a busy loop\n",
    "    log('future.result():', f.result())\n",
    "except: # notebook should always run all cells\n",
    "    print ('skipping this; pip install trollius required')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== from_marbles ==========\n",
      "\n",
      "module rx.testing.marbles\n",
      "@extensionclassmethod(Observable, alias=\"from_string\")\n",
      "def from_marbles(cls, string, scheduler=None):\n",
      "    Convert a marble diagram string to an observable sequence, using\n",
      "    an optional scheduler to enumerate the events.\n",
      "\n",
      "    Special characters:\n",
      "    - = Timespan of 100 ms\n",
      "    x = on_error()\n",
      "    | = on_completed()\n",
      "\n",
      "    All other characters are treated as an on_next() event at the given\n",
      "    moment they are found on the string.\n",
      "\n",
      "    Examples:\n",
      "    1 - res = rx.Observable.from_string(\"1-2-3-|\")\n",
      "    2 - res = rx.Observable.from_string(\"1-(42)-3-|\")\n",
      "    3 - res = rx.Observable.from_string(\"1-2-3-x\", rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    string -- String with marble diagram\n",
      "    scheduler -- [Optional] Scheduler to run the the input sequence on.\n",
      "\n",
      "    Returns the observable sequence whose elements are pulled from the\n",
      "    given marble diagram string.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   4.2     M New subscription on stream 276651201\n",
      "  16.2   T21 [next]   12.0: 1\n",
      " 127.7   T22 [next]  123.5: 42\n",
      " 236.9   T25 [next]  232.7: 3\n",
      " 347.0   T27 [cmpl]  342.7: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.from_marbles)\n",
    "d = subs(rx.Observable.from_string(\"1-(42)-3-|\").to_blocking())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== to_set ==========\n",
      "\n",
      "module rx.linq.observable.toset\n",
      "@extensionmethod(Observable)\n",
      "def to_set(self):\n",
      "    Converts the observable sequence to a set.\n",
      "\n",
      "    Returns {Observable} An observable sequence with a single value of a set\n",
      "    containing the values from the observable sequence.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.3     M New subscription on stream 276654145\n",
      "   3.0     M [next]    1.5: set(['a', 'c', 'b'])\n",
      "   3.4     M [cmpl]    1.9: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.to_set)\n",
    "d = subs(O.from_(\"abcabc\").to_set())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# I want an operator to operate on a particular Scheduler: **[subscribe_on](http://reactivex.io/documentation/operators/subscribeon.html)**\n",
    "\n",
    "Advanced feature: Adding **side effects** to subscription and unsubscription events.\n",
    "\n",
    "[This](http://tomstechnicalblog.blogspot.de/2016/02/rxjava-understanding-observeon-and.html) is a good read:\n",
    "\n",
    "![](https://i.imgur.com/CRLzESV.png)\n",
    "\n",
    "Plus see the other\n",
    "links on [RX docu](http://reactivex.io/documentation/operators/subscribeon.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== subscribe_on ==========\n",
      "\n",
      "module rx.linq.observable.subscribeon\n",
      "@extensionmethod(Observable)\n",
      "def subscribe_on(self, scheduler):\n",
      "    Subscribe on the specified scheduler.\n",
      "\n",
      "    Wrap the source sequence in order to run its subscription and\n",
      "    unsubscription logic on the specified scheduler. This operation is not\n",
      "    commonly used; see the remarks section for more information on the\n",
      "    distinction between subscribe_on and observe_on.\n",
      "\n",
      "    Keyword arguments:\n",
      "    scheduler -- Scheduler to perform subscription and unsubscription\n",
      "        actions on.\n",
      "\n",
      "    Returns the source sequence whose subscriptions and unsubscriptions\n",
      "    happen on the specified scheduler.\n",
      "\n",
      "    This only performs the side-effects of subscription and unsubscription\n",
      "    on the specified scheduler. In order to invoke observer callbacks on a\n",
      "    scheduler, use observe_on.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "\n",
      "========== Switching Schedulers ==========\n",
      "\n",
      "   3.6  T451 [next]    0.9: 42 (SimpleSubs)\n",
      "\n",
      "   2.6     M New subscription on stream 276546517   3.8  T451 [cmpl]    1.2: fin (SimpleSubs)\n",
      "\n",
      "\n",
      "\n",
      "========== Custom Subscription Side Effects ==========\n",
      "\n",
      "\n",
      " 106.6     M New subscription on stream 276546553\n",
      " 106.9     M new scheduling task <function action at 0x107fad5f0>\n",
      "\n",
      " 107.9     M New subscription on stream 276546553\n",
      " 108.5     M new scheduling task <function action at 0x107b82320>\n",
      " 310.2  T454 [next]  203.5: 0 (subs1)\n",
      " 311.6  T456 [next]  203.4: 0 (subs2)\n",
      " 513.5  T457 [next]  406.8: 1 (subs1)\n",
      " 515.4  T458 [next]  407.3: 1 (subs2) 513.8  T457 [cmpl]  407.2: fin (subs1)\n",
      " 516.2  T458 [cmpl]  408.1: fin (subs2)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "rst(O.subscribe_on)\n",
    "\n",
    "# start simple:\n",
    "header('Switching Schedulers')\n",
    "s = O.just(42, rx.concurrency.ImmediateScheduler())\n",
    "d = subs(s.subscribe_on(rx.concurrency.TimeoutScheduler()), name='SimpleSubs')\n",
    "\n",
    "sleep(0.1)\n",
    "\n",
    "header('Custom Subscription Side Effects')\n",
    "\n",
    "from rx.concurrency.newthreadscheduler import NewThreadScheduler\n",
    "from rx.concurrency.eventloopscheduler import EventLoopScheduler\n",
    "\n",
    "class MySched(NewThreadScheduler):\n",
    "    '''For adding side effects at subscription and unsubscription time'''\n",
    "    def schedule(self, action, state=None):\n",
    "        log('new scheduling task', action)\n",
    "        scheduler = EventLoopScheduler(\n",
    "            thread_factory=self.thread_factory,\n",
    "            exit_if_empty=True)\n",
    "        return scheduler.schedule(action, state)\n",
    "                \n",
    "s = O.interval(200).take(2)\n",
    "s = s.subscribe_on(MySched())\n",
    "d = subs(s, name=\"subs1\")\n",
    "d = subs(s, name=\"subs2\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## ...when it notifies observers **[observe_on](http://reactivex.io/documentation/operators/subscribeon.html)**\n",
    "\n",
    "Via this you can add side effects on any notification to any subscriber.\n",
    "\n",
    "This example shall demonstrate whats going on:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== observe_on ==========\n",
      "\n",
      "module rx.linq.observable.observeon\n",
      "@extensionmethod(Observable)\n",
      "def observe_on(self, scheduler):\n",
      "    Wraps the source sequence in order to run its observer callbacks on\n",
      "    the specified scheduler.\n",
      "\n",
      "    Keyword arguments:\n",
      "    scheduler -- Scheduler to notify observers on.\n",
      "\n",
      "    Returns the source sequence whose observations happen on the specified\n",
      "    scheduler.\n",
      "\n",
      "    This only invokes observer callbacks on a scheduler. In case the\n",
      "    subscription and/or unsubscription actions have side-effects\n",
      "    that require to be run on a scheduler, use subscribe_on.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "\n",
      "========== defining a custom thread factory for a custom scheduler ==========\n",
      "\n",
      "\n",
      "   2.8     M New subscription on stream 276529577\n",
      " 208.2  T392 RX called schedule on mysched\n",
      "\n",
      "\n",
      "created Thread-393\n",
      "\n",
      " 209.5  T393 [next]  206.4: 0\n",
      " 210.2  T393 RX called schedule on mysched\n",
      "\n",
      "\n",
      "created Thread-395\n",
      "\n",
      " 414.0  T394 RX called schedule on mysched\n",
      " 415.5  T396 [next]  412.4: 1\n",
      "\n",
      "\n",
      "created Thread-396\n",
      " 416.2  T396 RX called schedule on mysched\n",
      "\n",
      "\n",
      "\n",
      "created Thread-398\n",
      "\n",
      " 619.0  T397 RX called schedule on mysched\n",
      " 620.3  T399 [next]  617.3: 2\n",
      "\n",
      "\n",
      "created Thread-399\n",
      " 620.5  T399 RX called schedule on mysched\n",
      "\n",
      "\n",
      " 622.4  T400 [cmpl]  619.4: fin\n",
      "created Thread-400\n",
      "\n",
      "\n",
      " 622.9  T400 RX called schedule on mysched\n",
      "\n",
      "\n",
      "created Thread-402\n",
      "\n",
      "all threads after finish:\n",
      "MainThread Thread-2 IPythonHistorySavingThread Thread-1 Thread-3\n"
     ]
    }
   ],
   "source": [
    "rst(O.observe_on)\n",
    "from rx.concurrency.newthreadscheduler import NewThreadScheduler\n",
    "\n",
    "header('defining a custom thread factory for a custom scheduler')\n",
    "def my_thread_factory(target, args=None): \n",
    "    'just to show that also here we can customize'\n",
    "    t = threading.Thread(target=target, args=args or [])\n",
    "    t.setDaemon(True)\n",
    "    print ('\\ncreated %s\\n' % t.getName())\n",
    "    return t\n",
    "\n",
    "\n",
    "class MySched:\n",
    "    def __init__(self):\n",
    "        self.rx_sched = NewThreadScheduler(my_thread_factory)\n",
    "        \n",
    "    def __getattr__(self, a):\n",
    "        'called whenever the observe_on scheduler is on duty'\n",
    "        log('RX called', a, 'on mysched\\n')\n",
    "        return getattr(self.rx_sched, a)\n",
    "        \n",
    "mysched = MySched()        \n",
    "s = O.interval(200).take(3) #.delay(100, mysched)\n",
    "\n",
    "d = subs(s.observe_on(mysched))\n",
    "\n",
    "sleep(2)\n",
    "print 'all threads after finish:' # all cleaned up\n",
    "print (' '.join([t.name for t in threading.enumerate()]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# I want an Observable to invoke a particular action when certain events occur **[do_action/tap, finally_action](http://reactivex.io/documentation/operators/do.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== do_action ==========\n",
      "\n",
      "module rx.linq.observable.doaction\n",
      "@extensionmethod(Observable, alias=\"tap\")\n",
      "def do_action(self, on_next=None, on_error=None, on_completed=None,\n",
      "    Invokes an action for each element in the observable sequence and\n",
      "    invokes an action upon graceful or exceptional termination of the\n",
      "    observable sequence. This method can be used for debugging, logging,\n",
      "    etc. of query behavior by intercepting the message stream to run\n",
      "    arbitrary actions for messages on the pipeline.\n",
      "\n",
      "    1 - observable.do_action(observer)\n",
      "    2 - observable.do_action(on_next)\n",
      "    3 - observable.do_action(on_next, on_error)\n",
      "    4 - observable.do_action(on_next, on_error, on_completed)\n",
      "\n",
      "    observer -- [Optional] Observer, or ...\n",
      "    on_next -- [Optional] Action to invoke for each element in the\n",
      "        observable sequence.\n",
      "    on_error -- [Optional] Action to invoke upon exceptional termination\n",
      "        of the observable sequence.\n",
      "    on_completed -- [Optional] Action to invoke upon graceful termination\n",
      "        of the observable sequence.\n",
      "\n",
      "    Returns the source sequence with the side-effecting behavior applied.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.4     M New subscription on stream 276579277\n",
      "   4.1     M NI! 10\n",
      "   4.5     M [next]    1.0: 10\n",
      "   4.7     M NI! 11\n",
      "   4.9     M [next]    1.5: 11\n",
      "   5.1     M EOF\n",
      "   5.2     M [cmpl]    1.8: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.do_action)\n",
    "def say(v=None):\n",
    "    if v:\n",
    "       log('NI!', v)\n",
    "    else:\n",
    "       log('EOF')\n",
    "    \n",
    "d = subs(O.range(10, 10).take(2).tap(say, on_completed=say))    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== finally_action ==========\n",
      "\n",
      "module rx.linq.observable.finallyaction\n",
      "@extensionmethod(Observable)\n",
      "def finally_action(self, action):\n",
      "    Invokes a specified action after the source observable sequence\n",
      "    terminates gracefully or exceptionally.\n",
      "\n",
      "    Example:\n",
      "    res = observable.finally(lambda: print('sequence ended')\n",
      "\n",
      "    Keyword arguments:\n",
      "    action -- {Function} Action to invoke after the source observable\n",
      "        sequence terminates.\n",
      "    Returns {Observable} Source sequence with the action-invoking\n",
      "    termination behavior applied.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.2     M New subscription on stream 276553221\n",
      "   3.0     M [err ]    0.6: err\n",
      "   3.5     M EOF\n"
     ]
    }
   ],
   "source": [
    "rst(O.finally_action)\n",
    "d = subs(O.throw('err').take(2).finally_action(say))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# I want an Observable that will notify observers of an error [throw](http://reactivex.io/documentation/operators/throw.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== throw ==========\n",
      "\n",
      "module rx.linq.observable.throw\n",
      "@extensionclassmethod(Observable, alias=\"throw_exception\")\n",
      "def throw(cls, exception, scheduler=None):\n",
      "    Returns an observable sequence that terminates with an exception,\n",
      "    using the specified scheduler to send out the single OnError message.\n",
      "\n",
      "    1 - res = rx.Observable.throw_exception(Exception('Error'))\n",
      "    2 - res = rx.Observable.throw_exception(Exception('Error'),\n",
      "                                            rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    exception -- An object used for the sequence's termination.\n",
      "    scheduler -- Scheduler to send the exceptional termination call on. If\n",
      "        not specified, defaults to ImmediateScheduler.\n",
      "\n",
      "    Returns the observable sequence that terminates exceptionally with the\n",
      "    specified exception object.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.9     M New subscription on stream 276578025\n",
      "   2.6     M [next]    0.6: 1\n",
      "   3.0     M [next]    0.9: 2\n",
      "   3.3     M [next]    1.2: 3\n",
      "   3.6     M [err ]    1.5: ups\n"
     ]
    }
   ],
   "source": [
    "rst(O.throw)\n",
    "d = subs(O.range(1, 3).concat(O.throw(\"ups\")))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...if a specified period of time elapses without it emitting an item [timeout / timeout_with_selector](http://reactivex.io/documentation/operators/timeout.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== timeout ==========\n",
      "\n",
      "module rx.linq.observable.timeout\n",
      "@extensionmethod(Observable)\n",
      "def timeout(self, duetime, other=None, scheduler=None):\n",
      "    Returns the source observable sequence or the other observable\n",
      "    sequence if duetime elapses.\n",
      "\n",
      "    1 - res = source.timeout(new Date()); # As a date\n",
      "    2 - res = source.timeout(5000); # 5 seconds\n",
      "    # As a date and timeout observable\n",
      "    3 - res = source.timeout(datetime(), rx.Observable.return_value(42))\n",
      "    # 5 seconds and timeout observable\n",
      "    4 - res = source.timeout(5000, rx.Observable.return_value(42))\n",
      "    # As a date and timeout observable\n",
      "    5 - res = source.timeout(datetime(), rx.Observable.return_value(42),\n",
      "                             rx.Scheduler.timeout)\n",
      "    # 5 seconds and timeout observable\n",
      "    6 - res = source.timeout(5000, rx.Observable.return_value(42),\n",
      "                             rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    :param datetime|int duetime: Absolute (specified as a datetime object) or\n",
      "        relative time (specified as an integer denoting milliseconds) when a\n",
      "        timeout occurs.\n",
      "    :param Observable other: Sequence to return in case of a timeout. If not\n",
      "        specified, a timeout error throwing sequence will be used.\n",
      "    :param Scheduler scheduler: Scheduler to run the timeout timers on. If not\n",
      "        specified, the timeout scheduler is used.\n",
      "\n",
      "    :returns: The source sequence switching to the other sequence in case of\n",
      "        a timeout.\n",
      "    :rtype: Observable\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   4.2     M New subscription on stream 276577449\n",
      "  17.1  T521 [next]   12.8: a\n",
      " 127.8  T522 [next]  123.5: b\n",
      " 332.7  T530 [next]  328.4: timeout\n",
      " 333.0  T530 [cmpl]  328.7: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.timeout)\n",
    "d = subs(marble_stream(\"a-b---c|\").timeout(200, O.just('timeout')))\n",
    "# this also works with absolute time. See docstring:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== timeout_with_selector ==========\n",
      "\n",
      "module rx.linq.observable.timeoutwithselector\n",
      "@extensionmethod(Observable)\n",
      "def timeout_with_selector(self, first_timeout=None,\n",
      "                          timeout_duration_selector=None, other=None):\n",
      "    Returns the source observable sequence, switching to the other\n",
      "    observable sequence if a timeout is signaled.\n",
      "\n",
      "    1 - res = source.timeout_with_selector(rx.Observable.timer(500))\n",
      "    2 - res = source.timeout_with_selector(rx.Observable.timer(500),\n",
      "                lambda x: rx.Observable.timer(200))\n",
      "    3 - res = source.timeout_with_selector(rx.Observable.timer(500),\n",
      "                lambda x: rx.Observable.timer(200)),\n",
      "                rx.Observable.return_value(42))\n",
      "\n",
      "    first_timeout -- [Optional] Observable sequence that represents the\n",
      "        timeout for the first element. If not provided, this defaults to\n",
      "        Observable.never().\n",
      "    timeout_Duration_selector -- [Optional] Selector to retrieve an\n",
      "        observable sequence that represents the timeout between the current\n",
      "        element and the next element.\n",
      "    other -- [Optional] Sequence to return in case of a timeout. If not\n",
      "        provided, this is set to Observable.throw_exception().\n",
      "\n",
      "    Returns the source sequence switching to the other sequence in case of\n",
      "    a timeout.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.9     M New subscription on stream 276596365\n",
      "  17.0   T74 [next]   12.9: 2\n",
      " 126.1   T76 [next]  122.1: 2\n",
      " 237.1   T77 [next]  233.1: 1\n",
      " 341.6   T86 [next]  337.5: timeout\n",
      " 341.9   T86 [cmpl]  337.8: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.timeout_with_selector)\n",
    "d = subs(marble_stream(\"2-2-1-1|\")\\\n",
    "         .timeout_with_selector(\n",
    "              # you get the value and can adjust the timeout accordingly:\n",
    "              timeout_duration_selector=lambda x: O.timer(100 * int(x)),\n",
    "              other=O.just('timeout')))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# I want an Observable to recover gracefully"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...from a timeout by switching to a backup Observable [timeout / timeout_with_selector](http://reactivex.io/documentation/operators/timeout.html)**\n",
    "(example: see above)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## ...from an upstream error notification [catch_exception, on_error_resume_next](http://reactivex.io/documentation/operators/catch.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== catch_exception ==========\n",
      "\n",
      "module rx.linq.observable.catch\n",
      "@extensionclassmethod(Observable)\n",
      "def catch_exception(cls, *args):\n",
      "    Continues an observable sequence that is terminated by an\n",
      "    exception with the next observable sequence.\n",
      "\n",
      "    1 - res = Observable.catch_exception(xs, ys, zs)\n",
      "    2 - res = Observable.catch_exception([xs, ys, zs])\n",
      "\n",
      "    Returns an observable sequence containing elements from consecutive\n",
      "    source sequences until a source sequence terminates successfully.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   3.5     M New subscription on stream 276515289\n",
      "   4.5     M [next]    0.8: 42\n",
      "   4.6     M [cmpl]    1.0: fin\n",
      "\n",
      "\n",
      "========== on_error_resume_next ==========\n",
      "\n",
      "module rx.linq.observable.onerrorresumenext\n",
      "@extensionclassmethod(Observable)\n",
      "def on_error_resume_next(cls, *args):\n",
      "    Continues an observable sequence that is terminated normally or by\n",
      "    an exception with the next observable sequence.\n",
      "\n",
      "    1 - res = Observable.on_error_resume_next(xs, ys, zs)\n",
      "    2 - res = Observable.on_error_resume_next([xs, ys, zs])\n",
      "    3 - res = Observable.on_error_resume_next(xs, factory)\n",
      "\n",
      "    Returns an observable sequence that concatenates the source sequences,\n",
      "    even if a sequence terminates exceptionally.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.2     M New subscription on stream 276515085\n",
      "   2.6     M [next]    0.3: running\n",
      "   2.8     M [next]    0.5: 0\n",
      "   3.0     M [next]    0.7: 1\n",
      "   3.1     M [next]    0.8: 2\n",
      "   3.2     M [next]    0.9: 3\n",
      "   3.3     M notify complete\n",
      "   3.8     M [next]    1.6: 0\n",
      "   4.3     M [next]    2.0: 1\n",
      "   4.5     M [next]    2.2: 2\n",
      "   4.7     M [next]    2.4: 3\n",
      "   5.0     M [next]    2.7: 4\n",
      "   5.2     M notify error\n",
      "   5.4     M notify complete\n",
      "   6.0     M [next]    3.8: all good\n",
      "   6.4     M [cmpl]    4.2: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.catch_exception)\n",
    "fubar1 = O.throw('Ups')\n",
    "fubar2 = O.throw('Argh')\n",
    "good = O.just(42)\n",
    "d = subs(O.catch_exception(fubar1, fubar2, good))\n",
    "\n",
    "\n",
    "rst(O.on_error_resume_next)\n",
    "\n",
    "bucket = [0]\n",
    "def emitter(obs):\n",
    "    v = bucket[-1]\n",
    "    bucket.append(v)\n",
    "    for i in range(0, len(bucket) + 2):\n",
    "        obs.on_next(i)\n",
    "    if len(bucket) > 2:\n",
    "        log('notify error')\n",
    "        obs.on_error(\"ups\")\n",
    "    log('notify complete')\n",
    "    obs.on_completed()\n",
    "    \n",
    "    \n",
    "    \n",
    "d = subs(O.on_error_resume_next(O.just('running'),\n",
    "                                O.create(emitter),\n",
    "                                O.create(emitter),\n",
    "                                O.just('all good')\n",
    "                               ))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ... by attempting to resubscribe to the upstream Observable **[retry](http://reactivex.io/documentation/operators/retry.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== retry ==========\n",
      "\n",
      "module rx.linq.observable.retry\n",
      "@extensionmethod(Observable)\n",
      "def retry(self, retry_count=None):\n",
      "    Repeats the source observable sequence the specified number of times\n",
      "    or until it successfully terminates. If the retry count is not\n",
      "    specified, it retries indefinitely.\n",
      "\n",
      "    1 - retried = retry.repeat()\n",
      "    2 - retried = retry.repeat(42)\n",
      "\n",
      "    retry_count -- [Optional] Number of times to retry the sequence. If not\n",
      "    provided, retry the sequence indefinitely.\n",
      "\n",
      "    Returns an observable sequence producing the elements of the given\n",
      "    sequence repeatedly until it terminates successfully.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   1.4     M New subscription on stream 276578989\n",
      "   1.8     M [next]    0.3: try 0.00107383728027\n",
      " 204.6     M error\n",
      " 205.8     M [next]  204.3: try 0.205071926117\n",
      " 408.9     M error\n",
      " 410.7     M [next]  409.2: try 0.409952878952\n",
      " 613.3     M error\n",
      " 614.7     M [next]  613.2: try 0.613961935043\n",
      " 819.0     M error\n",
      " 820.0     M [next]  818.4: try 0.819205999374\n",
      "1024.7     M error\n",
      "1025.2     M [next] 1023.7: try 1.02444982529\n",
      "1025.4     M [cmpl] 1023.9: fin\n"
     ]
    }
   ],
   "source": [
    "rst(O.retry)\n",
    "ts = time.time()\n",
    "def emit(obs):\n",
    "    dt = time.time() - ts\n",
    "    obs.on_next('try %s' % dt)\n",
    "    if dt < 1:\n",
    "        sleep(0.2)\n",
    "        log('error')\n",
    "        obs.on_error('ups')\n",
    "    obs.on_completed()\n",
    "    \n",
    "d = subs(O.create(emit).retry(10))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# I want to create a resource that has the same lifespan as the Observable **[using](http://reactivex.io/documentation/operators/using.html)**\n",
    "\n",
    "http://www.introtorx.com/Content/v1.0.10621.0/11_AdvancedErrorHandling.html#Using:\n",
    "\n",
    "The Using factory method allows you to bind the lifetime of a resource to the lifetime of an observable sequence. The signature itself takes two factory methods; one to provide the resource and one to provide the sequence. This allows everything to be lazily evaluated.\n",
    "\n",
    "This mechanism can find varied practical applications in the hands of an imaginative developer. The resource being an IDisposable is convenient; indeed, it makes it so that many types of resources can be bound to, such as other subscriptions, stream reader/writers, database connections, user controls and, with Disposable.Create(Action), virtually anything else."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== using ==========\n",
      "\n",
      "module rx.linq.observable.using\n",
      "@extensionclassmethod(Observable)\n",
      "def using(cls, resource_factory, observable_factory):\n",
      "    Constructs an observable sequence that depends on a resource object,\n",
      "    whose lifetime is tied to the resulting observable sequence's lifetime.\n",
      "\n",
      "    1 - res = rx.Observable.using(lambda: AsyncSubject(), lambda: s: s)\n",
      "\n",
      "    Keyword arguments:\n",
      "    resource_factory -- Factory function to obtain a resource object.\n",
      "    observable_factory -- Factory function to obtain an observable sequence\n",
      "        that depends on the obtained resource.\n",
      "\n",
      "    Returns an observable sequence whose lifetime controls the lifetime of\n",
      "    the dependent resource object.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.2     M New subscription (outer stream) on stream 276622297\n",
      "\n",
      "   2.6     M New subscription (resource fac) on stream 276622093\n",
      " 106.8  T127 [next]  104.1: 0 -> resource fac\n",
      "  \n",
      " 213.1  T129 [next]  210.5: 1 -> resource fac\n",
      "  \n",
      " 318.8  T130 [next]  316.2: 2 -> resource fac\n",
      "  \n",
      " 420.2  T131 [next]  417.5: 3 -> resource fac\n",
      "  \n",
      " 521.3  T132 [next]  518.7: 4 -> resource fac\n",
      "  \n",
      " 622.3  T133 [next]  619.7: 5 -> resource fac\n",
      "  \n",
      " 724.3  T134 [next]  721.7: 6 -> resource fac\n",
      "  \n",
      " 828.5  T135 [next]  825.8: 7 -> resource fac\n",
      "  \n",
      " 929.4  T136 [next]  926.7: 8 -> resource fac\n",
      "  \n",
      "1031.6  T137 [next] 1028.9: 9 -> resource fac\n",
      "  \n",
      "1133.8  T138 [next] 1131.1: 10 -> resource fac\n",
      "  \n",
      "1239.0  T139 [next] 1236.4: 11 -> resource fac\n",
      "  \n",
      "1342.8  T140 [next] 1340.1: 12 -> resource fac\n",
      "  \n",
      "1446.2  T141 [next] 1443.5: 13 -> resource fac\n",
      "  \n",
      "1550.1  T142 [next] 1547.5: 14 -> resource fac\n",
      "  \n",
      "1655.1  T143 [next] 1652.4: 15 -> resource fac\n",
      "  \n",
      "1757.7  T144 [next] 1755.0: 16 -> resource fac\n",
      "  \n",
      "1858.7  T145 [next] 1856.1: 17 -> resource fac\n",
      "  \n",
      "1961.2  T146 [next] 1958.5: 18 -> resource fac\n",
      "  \n",
      "2005.6  T128 [next] 2003.4: <rx.disposables.anonymousdisposable.AnonymousDisposable object at 0x107d04450> -> outer stream\n",
      "  \n",
      "2006.0  T128 [cmpl] 2003.7: fin -> outer stream\n",
      "  \n"
     ]
    }
   ],
   "source": [
    "rst(O.using)\n",
    "#d = subs(O.interval(1000).take(2))\n",
    "\n",
    "lifetime = 2000\n",
    "def result(disposable_resource_fac):\n",
    "    return O.just(disposable_resource_fac).delay(lifetime)\n",
    "\n",
    "d2 = subs(O.using(lambda: subs(O.interval(100).take(1000), name='resource fac\\n'),\n",
    "                  result), name='outer stream\\n')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# I want to subscribe to an Observable and receive a Future that blocks until the Observable completes **[start, start_async, to_async](http://reactivex.io/documentation/operators/start.html)**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== start ==========\n",
      "\n",
      "module rx.linq.observable.start\n",
      "@extensionclassmethod(Observable)\n",
      "def start(cls, func, scheduler=None):\n",
      "    Invokes the specified function asynchronously on the specified\n",
      "    scheduler, surfacing the result through an observable sequence.\n",
      "\n",
      "    Example:\n",
      "    res = rx.Observable.start(lambda: pprint('hello'))\n",
      "    res = rx.Observable.start(lambda: pprint('hello'), rx.Scheduler.timeout)\n",
      "\n",
      "    Keyword arguments:\n",
      "    func -- {Function} Function to run asynchronously.\n",
      "    scheduler -- {Scheduler} [Optional] Scheduler to run the function on. If\n",
      "        not specified, defaults to Scheduler.timeout.\n",
      "\n",
      "    Returns {Observable} An observable sequence exposing the function's\n",
      "    result value, or an exception.\n",
      "\n",
      "    Remarks:\n",
      "    The function is called immediately, not during the subscription of the\n",
      "    resulting sequence. Multiple subscriptions to the resulting sequence can\n",
      "    observe the function's result.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.1     M New subscription (sub1) on stream 276594277\n",
      "   2.6     M [next]    0.2: ('start: ', 1482383023.989834) -> sub1  \n",
      "   2.9     M [next]    0.6: a -> sub1  \n",
      "   3.1     M [next]    0.7: b -> sub1  \n",
      "   3.2     M [next]    0.8: c -> sub1  \n",
      "   3.5     M [cmpl]    1.1: fin -> sub1  \n",
      "\n",
      "   3.9     M New subscription (sub2) on stream 276594277\n",
      "   4.2     M [next]    0.2: ('start: ', 1482383023.989834) -> sub2  \n",
      "   4.6     M [next]    0.6: a -> sub2  \n",
      "   4.9     M [next]    1.0: b -> sub2  \n",
      "   5.1     M [next]    1.2: c -> sub2  \n",
      "   5.3     M [cmpl]    1.3: fin -> sub2  \n"
     ]
    }
   ],
   "source": [
    "rst(O.start)\n",
    "def starter():\n",
    "    # called only once, async:\n",
    "    return 'start: ', time.time()\n",
    "s = O.start(starter).concat(O.from_('abc'))\n",
    "d = subs(s, name='sub1')\n",
    "d = subs(s, name='sub2')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== start_async ==========\n",
      "\n",
      "module rx.linq.observable.startasync\n",
      "@extensionclassmethod(Observable)\n",
      "def start_async(cls, function_async):\n",
      "    Invokes the asynchronous function, surfacing the result through an\n",
      "    observable sequence.\n",
      "\n",
      "    Keyword arguments:\n",
      "    :param types.FunctionType function_async: Asynchronous function which\n",
      "        returns a Future to run.\n",
      "\n",
      "    :returns: An observable sequence exposing the function's result value, or an\n",
      "        exception.\n",
      "    :rtype: Observable\n",
      "--------------------------------------------------------------------------------\n",
      "   1.8     M called future\n",
      "\n",
      "   2.1     M New subscription (subs1) on stream 276008249\n",
      "   2.9     M [err ]    0.2: FINISHED: <Future finished result=('42', 1482384109.227631)> -> subs1  \n",
      "\n",
      "   3.5     M New subscription (subs2) on stream 276008249\n",
      "   3.9     M [err ]    0.1: FINISHED: <Future finished result=('42', 1482384109.227631)> -> subs2  \n"
     ]
    }
   ],
   "source": [
    "rst(O.start_async)\n",
    "\n",
    "def emit(obs):\n",
    "    \n",
    "    for ev in 'first', 'second':\n",
    "        sleep(.2)\n",
    "        log('emitting', ev)\n",
    "        obs.on_next(ev)\n",
    "    # vital for the future to get done:\n",
    "    obs.on_completed()\n",
    "    \n",
    "def future():\n",
    "    # only called once:\n",
    "    log('called future')\n",
    "    future = trollius.Future()\n",
    "    future.set_result(('42', time.time()))\n",
    "    future.set_exception(Exception('ups'))\n",
    "    return future\n",
    "    \n",
    "try:\n",
    "    # required for py2 (backport of guidos' tulip stuffs, now asyncio)\n",
    "    # caution: people say this is not production ready and will never be.\n",
    "    import trollius\n",
    "    s = O.start_async(future)\n",
    "    d = subs(s, name='subs1')\n",
    "    # same result:\n",
    "    d = subs(s, name='subs2')\n",
    "except Exception as ex: # notebook should always run all cells\n",
    "    print ('%s skipping this; pip install trollius required' % ex)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "========== to_async ==========\n",
      "\n",
      "module rx.linq.observable.toasync\n",
      "@extensionclassmethod(Observable)\n",
      "def to_async(cls, func, scheduler=None):\n",
      "    Converts the function into an asynchronous function. Each invocation\n",
      "    of the resulting asynchronous function causes an invocation of the\n",
      "    original synchronous function on the specified scheduler.\n",
      "\n",
      "    Example:\n",
      "    res = Observable.to_async(lambda x, y: x + y)(4, 3)\n",
      "    res = Observable.to_async(lambda x, y: x + y, Scheduler.timeout)(4, 3)\n",
      "    res = Observable.to_async(lambda x: log.debug(x),\n",
      "                              Scheduler.timeout)('hello')\n",
      "\n",
      "    func -- {Function} Function to convert to an asynchronous function.\n",
      "    scheduler -- {Scheduler} [Optional] Scheduler to run the function on. If\n",
      "        not specified, defaults to Scheduler.timeout.\n",
      "\n",
      "    Returns {Function} Asynchronous function.\n",
      "--------------------------------------------------------------------------------\n",
      "\n",
      "   2.9     M New subscription (9223) on stream 276564689\n",
      "   3.4     M [next]    0.4: 7 -> 9223  \n",
      "   3.6     M [cmpl]    0.5: fin -> 9223  \n"
     ]
    }
   ],
   "source": [
    "rst(O.to_async)\n",
    "d = subs(O.to_async(lambda x, y: x + y)(4, 3) )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
